package com.hung.trade.common.rocketmq;

import com.hung.trade.common.constants.TradeEnums;
import com.hung.trade.common.exception.TradeException;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author hungki
 * @createTime 2018-10-07 下午 11:55
 */
public class RocketMQProducer {
    private static Logger logger = LoggerFactory.getLogger(RocketMQProducer.class);

    public String getGroupName() {
        return groupName;
    }

    public void setGroupName(String groupName) {
        this.groupName = groupName;
    }

    public String getNamesvrAddr() {
        return namesvrAddr;
    }

    public void setNamesvrAddr(String namesvrAddr) {
        this.namesvrAddr = namesvrAddr;
    }

    public int getMaxMessageSize() {
        return maxMessageSize;
    }

    public void setMaxMessageSize(int maxMessageSize) {
        this.maxMessageSize = maxMessageSize;
    }

    public int getSendMsgTimeout() {
        return sendMsgTimeout;
    }

    public void setSendMsgTimeout(int sendMsgTimeout) {
        this.sendMsgTimeout = sendMsgTimeout;
    }

    private String groupName;
    private String namesvrAddr;
    private int maxMessageSize = 1024 * 1024 *3;
    private int sendMsgTimeout  =3000;
    private DefaultMQProducer producer;

    public void init() throws TradeException {
        producer = new DefaultMQProducer(groupName);
        producer.setNamesrvAddr(namesvrAddr);
        producer.setMaxMessageSize(maxMessageSize);
        producer.setSendMsgTimeout(sendMsgTimeout);
        try {
            producer.start();
            logger.info(String.format("producer is start! groupName:[%s],namesvrAddr:[%s]",groupName,namesvrAddr));
        } catch (MQClientException e) {
            logger.error(String.format("producer error! groupName:[%s],namesvrAddr:[%s]",groupName,namesvrAddr),e);
            throw new TradeException(e);
        }
    }
    public SendResult sendMsg(String topic,String tags,String keys,String messageText) throws TradeException {
        if(StringUtils.isEmpty(topic)){
            logger.error(String.format("topic is empty"));
            throw new TradeException("topic is empty");
        }
        SendResult sendResult;
        Message message = new Message(topic,tags,keys,messageText.getBytes());
        try {
            sendResult =this.producer.send(message);
            logger.debug(String.format("message[topic = [%s],tags = [%s],keys = [%s]] send sucess",topic,tags,keys));
        } catch (MQClientException|RemotingException|MQBrokerException|InterruptedException e) {
            logger.debug(String.format("message[topic = [%s],tags = [%s],keys = [%s]] send error!",topic,tags,keys),e);
            throw new TradeException(e);
        }
        return sendResult;
    }
    public SendResult sendMsg(TradeEnums.OrderTopicAndTagEnum orderTopicAndTagEnum, String keys, String messageText) throws TradeException {

        SendResult sendResult;
        Message message = new Message(orderTopicAndTagEnum.getTopic(),orderTopicAndTagEnum.getTags(),keys,messageText.getBytes());
        try {
            sendResult =this.producer.send(message);
            logger.debug(String.format("message[topic = [%s],tags = [%s],keys = [%s]] send sucess",orderTopicAndTagEnum.getTopic(),orderTopicAndTagEnum.getTags(),keys));
        } catch (MQClientException|RemotingException|MQBrokerException|InterruptedException e) {
            logger.debug(String.format("message[topic = [%s],tags = [%s],keys = [%s]] send error!",orderTopicAndTagEnum.getTopic(),orderTopicAndTagEnum.getTags(),keys),e);
            throw new TradeException(e);
        }
        return sendResult;
    }
}
